/*
 * Copyright 2024 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.pajamas.example.client.starter.integration.module.client;

import com.rabbitmq.client.Channel;

import org.pajamas.daydream.serializer.json.Jsons;
import org.pajamas.example.client.starter.core.entity.AlbumEntity;
import org.pajamas.starter.integration.spring.amqp.Namings;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.handler.annotation.Payload;

import lombok.extern.slf4j.Slf4j;

/**
 * @author william
 * @since 2024/5/23
 */
@Slf4j
public class ExampleQueueClientConfiguration {
    static final String NAME_EXCHANGE = "test.exchange";
    static final String NAME_QUEUE = "test.Queue";
    static final String ROUTING_KEY = "test.routingKey";
    static final String HEADER_KEY_ACK = "test-ack";
    static final String HEADER_KEY_THROW = "test-throw";

    @Bean
    public Exchange testExchange() {
        return ExchangeBuilder
                .topicExchange(NAME_EXCHANGE)
                .durable(false)
                .autoDelete()
                .alternate(Namings.DLX)
                .build();
    }

    @Bean
    public Queue testQueue() {
        return QueueBuilder
                .nonDurable(NAME_QUEUE)
                .autoDelete()
                .deadLetterExchange(Namings.DLX)
                .deadLetterRoutingKey(Namings.DLRK)
                .ttl(1000)
                .build();
    }

    @Bean
    public Binding testBinding(@Qualifier("testExchange") Exchange exchange, @Qualifier("testQueue") Queue queue) {
        return BindingBuilder
                .bind(queue)
                .to(exchange)
                .with(ROUTING_KEY)
                .noargs();
    }

    @RabbitListener(queues = NAME_QUEUE)
    public void onSubscribeTestQueue(Message message, Channel channel, @Payload AlbumEntity entity)
            throws Exception {
        log.info("onSubscribeTestQueue.messageProperty:[{}], payload:[{}]"
                , message.getMessageProperties()
                , Jsons.PRETTY.stringify(entity));

        Boolean canAck = message.getMessageProperties().getHeader(HEADER_KEY_ACK);
        if (canAck != null && !canAck) {
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            channel.basicReject(deliveryTag, false);
        }


        Boolean canThrow = message.getMessageProperties().getHeader(HEADER_KEY_THROW);
        if (canThrow != null && canThrow) {
            throw new Exception("HEADER_KEY_THROW is true");
        }
    }
}
